package com.lzwk.app.ods.ocpx;

import com.lzwk.beans.ocpx.OdsPayOrder;
import com.lzwk.common.ocpx.OCPXConstant;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: FlinkCDC
 * @Author Yeaser
 * @Date: 2022/1/29 15:05
 * @Version 1.0
 * Description: TODO
 */
public class OdsLookUpJoinTables {

    public static SingleOutputStreamOperator<OdsPayOrder> ocpxPayOrder(StreamTableEnvironment tenv) {

        tenv.getConfig().getConfiguration().setString("pipeline.name","ocpx_ods_order");
         tenv.executeSql("create table mysql_pay_order(" +
                " `id` INT, " +
                " `payee_id` INT, " +
                " `payee_type` STRING, " +
                " `fee` INT, " +
                " `product_type` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `paid` INT, " +
                " `deliver_status` INT, " +
                " `refund_status` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `meta` STRING NOT NULL, " +
                " `ip` STRING, " +
                " `payer_id` INT, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCPAYORDER + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")");


        Table table = tenv.sqlQuery(" select " +
                " t1.`payer_id`, " +
                " t1.`product_id`, " +
                " ifnull(t1.`source_id`, 1) AS `source_id`, " +
                " ifnull(t1.`skin_id`, -1) AS `skin_id`, " +
                " ifnull(t1.`wxapp_id`, -1) AS `wxapp_id`, " +
                " ifnull(t7.`org_id`, -1) AS `org_id`, " +
                " ifnull(t5.`tag_id`, -1) AS `tag_id`, " +
                " ifnull(regexp_replace(t3.`name`,'\\n|\\t',''), '') AS `source_name`, " +
                " ifnull(regexp_replace(t4.`name`,'\\n|\\t',''), '') AS `skin_name`, " +
                " ifnull(regexp_replace(t5.`name`,'\\n|\\t',''), '') AS `product_name`, " +
                " ifnull(regexp_replace(t8.`tag_name`,'\\n|\\t',''), '') AS `tag_name`, " +
                " ifnull(regexp_replace(t6.`nickname`,'\\n|\\t',''), '') AS `wxapp_nickname`, " +
                " ifnull(t5.`src_prod_id`,-1) as src_prod_id, " +
                " ifnull(CAST(REGEXP_EXTRACT(GetJsonString(t1.`meta`, 'url'), 'creativeid=([0-9]+)', 1) AS STRING),'') AS creative_id, " +
                " ifnull(CAST(REGEXP_EXTRACT(GetJsonString(t1.`meta`, 'url'), 'wx_aid=([0-9]+)', 1) AS STRING),'') AS wx_aid, " +
                " ifnull(CAST(REGEXP_EXTRACT(GetJsonString(t1.`meta`, 'url'), 'ks_aid=([0-9]+)', 1) AS STRING)  ,'') AS ks_aid, " +
                " ifnull(GetJsonString(t1.`meta`, 'openid'),'') AS `open_id`, " +
                " if(t1.source='dyshop',GetJsonInt(t1.`meta`,'pay_amount'),t1.`fee`) AS `fee`, " +
                " ifnull(GetJsonString(t1.`meta`, 'method'),'') AS `method`, " +
                " t1.`payee_id`, " +
                " ifnull(regexp_replace(t1.`payee_type`,'\\n|\\t',''),'') AS `payee_type`, " +
                " ifnull(t1.`product_type`,-1) AS product_type, " +
                " t1.`paid`, " +
                " t1.`deliver_status`, " +
                " t1.`refund_status`, " +
                " ifnull(t1.`create_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `create_time`, " +
                " ifnull(t1.`pay_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `pay_time`, " +
                " ifnull(t1.`callback_time`,TO_TIMESTAMP('1970-01-01 08:00:01'))  AS `callback_time`, " +
                " ifnull(regexp_replace(t1.`outer_order_id`,'\\n|\\t',''),'') AS `outer_order_id`, " +
                " ifnull(regexp_replace(t1.`ip`,'\\n|\\t',''),'') AS `ip`, " +
                " ifnull(regexp_replace(t1.`source`,'\\n|\\t',''),'') AS `source`, " +
                " ifnull(t1.`is_notified`,-1) AS `is_notified`, " +
                " ifnull(t1.`update_time`, TO_TIMESTAMP('1970-01-01 08:00:01')) AS `update_time`, " +
                " ifnull(regexp_replace(t1.`openid`,'\\n|\\t',''),'') AS `openid`, " +
                " ifnull(regexp_replace(t1.`meta`,'\\n|\\t',''),'') AS `meta`, " +
                " t1.`id`, " +
                " ifnull(t3.`meta`,'') AS `platform_meta`, " +
                " ifnull(GetJsonString(t3.`meta`,'source_platform'),'') AS `source_platform`," +
                " `proc_time` " +
                " FROM " +
                "     ( " +
                "     select " +
                "         *, " +
                "         row_number() over (partition by `product_id`,`payer_id` order by paid desc) as rn " +
                "    from mysql_pay_order) t1 " +
                " LEFT JOIN ocpx_promote_promotesource FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                " ON t1.source_id=t3.id " +
                " LEFT JOIN ocpx_product_productskin FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                " ON t1.skin_id=t4.id " +
                " LEFT JOIN ocpx_product_product FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                " ON t1.product_id = t5.id " +
                " LEFT JOIN ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` t6 " +
                " ON t1.wxapp_id = t6.id " +
                " LEFT JOIN ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t7 " +
                " ON t1.product_id=t7.object_id " +
                " AND t7.object_type = 'product' " +
                " LEFT JOIN ocpx_tag_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                " ON t5.tag_id=t8.id " +
                " WHERE t5.src_prod_type = 'wk_act' and rn = 1  "
        );

        //table.execute().print();

        tenv.createTemporaryView("ods_ocpx_pay_order",table);

        SingleOutputStreamOperator<OdsPayOrder> res = tenv
                .toRetractStream(table, OdsPayOrder.class)
                .filter(data -> data.f0)
                .map(data -> data.f1);

        return  res;
    }

    // 双流join 拿取trace_id和wxapp_id -> 高沉浸度，暂时保留代码
    /*public static SingleOutputStreamOperator<OdsWxappMpadsubfeedbackdata> ocpxWxAppMPAdSubFeedbackData(StreamTableEnvironment tenv) {


        tenv.executeSql("create table mysql_wxapp_mpad_subfeedbackdatas(" +
                " `id` INT, " +
                " `trace_id` STRING, " +
                " `account_info_id` INT, " +
                " `openid` STRING, " +
                " `wechat_account_id` STRING, " +
                " `wechat_agency_id` STRING, " +
                " `campaign_id` STRING, " +
                " `ad_group_id` STRING, " +
                " `subscribe_time` STRING, " +
                " `position_id` STRING, " +
                " `create_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCMPFEEDBACK + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")");


        Table table = tenv.sqlQuery(" select " +
                " `t1`.`id`, " +
                " regexp_replace(`t1`.`trace_id`,'\\n|\\t','') as trace_id, " +
                " regexp_replace(`t1`.`openid`,'\\n|\\t','') as openid, " +
                " regexp_replace(`t1`.`wechat_account_id`,'\\n|\\t','') as wechat_account_id, " +
                " `t1`.`account_info_id`, " +
                " regexp_replace(`t1`.`wechat_agency_id`,'\\n|\\t','') as wechat_agency_id, " +
                " regexp_replace(`t1`.`campaign_id`,'\\n|\\t','') as campaign_id, " +
                " regexp_replace(`t1`.`ad_group_id`,'\\n|\\t','') as ad_group_id, " +
                " regexp_replace(`t1`.`subscribe_time`,'\\n|\\t','') as subscribe_time, " +
                " `t1`.`position_id`, " +
                " ifnull(t1.`create_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) as create_time, " +
                " `t1`.`update_time`, " +
                " `t2`.id as `wxapp_id` " +
                " from mysql_wxapp_mpad_subfeedbackdatas t1" +
                " left  join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                " on regexp_replace(`t1`.`wechat_account_id`,'\\n|\\t','') = `t2`.`appid` " +
                " where t2.id is not null");

        tenv.createTemporaryView("ods_ocpx_wxapp_mpadsubfeedbackdata",table);

        //table.execute().print();

        SingleOutputStreamOperator<OdsWxappMpadsubfeedbackdata> res = tenv
                .toRetractStream(table, OdsWxappMpadsubfeedbackdata.class)
                .filter(data -> data.f0)
                .map(data -> data.f1);

        return res;
    }*/
}
